package org.intrasearch.core.collections;

import org.intrasearch.core.io.*;

import java.io.*;
import java.util.*;

/**
 * PATISDG 13.11.2009
 */
public class BigMap<K, V> implements Map<K, V>
{
    private final File index;
    private TreeMap<K, Position> indexMap;
    private Position position;
    private Map<Integer,Integer> allocationsPerChunk;
    private final int chunkSizeThreshold;
    private final SerializationConfig serializationConfig;

    public BigMap(File index, int chunkSizeThreshold, final SerializationConfig serializationConfig)
    {
        this.index = index;
        this.chunkSizeThreshold = chunkSizeThreshold;
        this.serializationConfig = serializationConfig(serializationConfig);
        initFromFile();
    }

    private SerializationConfig serializationConfig(final SerializationConfig serializationConfig)
    {
        return new CachingSerializationConfig(new SerializationConfig()
        {
            public void configure(SerializationBuilder builder)
            {
                builder.addExternalizableProvider(Integer.class, new IntegerExternalizableProvider());
                builder.addExternalizableProvider(TreeMap.class, new MapExternalizableProvider(TreeMap.class));
                builder.addClass(Position.class);
                serializationConfig.configure(builder);
            }
        });
    }

    private void initFromFile()
    {
        if (index.exists() && index.length() != 0)
        {
            read();
        }
        else
        {
            indexMap = new TreeMap<K, Position>();
            position = new Position(0,0);
            allocationsPerChunk = new TreeMap<Integer, Integer>();
        }
    }

    public int size()
    {
        return indexMap.size();
    }

    public boolean isEmpty()
    {
        return indexMap.isEmpty();
    }

    public boolean containsKey(Object key)
    {
        return indexMap.containsKey(key);
    }

    public boolean containsValue(Object value)
    {
        //Not compiled if no explicit type
        return this.<Boolean>throwExpensiveOperation();
    }

    private <T> T throwExpensiveOperation()
    {
        throw new RuntimeException("Not implemented cause it very expensive operation.");
    }

    public V get(Object key)
    {
        FileInputStream fileInputStream = null;
        ObjectInputStream objectInputStream = null;
        try
        {
            Position valuePosition = indexMap.get(key);
            if (valuePosition == null)
            {
                return null;
            }
            fileInputStream = new FileInputStream(getChunkFileByPosition(valuePosition));
            fileInputStream.getChannel().position(valuePosition.inChunkPosition);
            objectInputStream = createObjectInputStream(fileInputStream);
            return (V) objectInputStream.readObject();
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }
        finally
        {
            closeStreams(fileInputStream, objectInputStream);
        }
    }

    private ObjectInputStream createObjectInputStream(FileInputStream fileInputStream)
            throws IOException
    {
        return new StraightExternalizableInputStream(fileInputStream, serializationConfig);
    }

    private ObjectOutputStream createObjectOutputStream(FileOutputStream fileOutputStream)
            throws IOException
    {
        return new StraightExternalizableOutputStream(fileOutputStream, serializationConfig);
    }

    private void closeStream(Closeable closeable)
    {
        if (closeable != null)
        {
            try
            {
                closeable.close();
            }
            catch (IOException e)
            {
                throw new RuntimeException(e);
            }
        }
    }

    public V put(K key, V value)
    {
        FileOutputStream fileOutputStream = null;
        ObjectOutputStream objectOutputStream = null;
        try
        {
            Position writePosition = getWritePosition();
            fileOutputStream = new FileOutputStream(getChunkFileByPosition(writePosition), true);
            fileOutputStream.getChannel().position(writePosition.inChunkPosition);
            putToIndex(key, writePosition);
            objectOutputStream = createObjectOutputStream(fileOutputStream);
            objectOutputStream.writeObject(value);
            objectOutputStream.flush();
            int chunkPosition = (int) fileOutputStream.getChannel().position();
            advanceWritePosition(chunkPosition);
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }
        finally
        {
            closeStreams(fileOutputStream, objectOutputStream);
        }
        return null;  //TODO
    }

    private void putToIndex(K key, Position writePosition)
    {
        Position oldPosition = indexMap.put(key, writePosition);
        Integer newPositionChunkAllocationsCount = allocationsPerChunk.get(writePosition.chunkIndex);
        if (newPositionChunkAllocationsCount == null)
        {
            newPositionChunkAllocationsCount = 0;
        }
        newPositionChunkAllocationsCount++;
        allocationsPerChunk.put(writePosition.chunkIndex, newPositionChunkAllocationsCount);
        if (oldPosition != null)
        {
            Integer oldPositionChunkAllocationsCount = allocationsPerChunk.get(oldPosition.chunkIndex) - 1;
            allocationsPerChunk.put(oldPosition.chunkIndex, oldPositionChunkAllocationsCount);
            if (oldPositionChunkAllocationsCount == 0)
            {
                collectGarbage(oldPosition.chunkIndex);
            }
        }
    }

    private void collectGarbage(int chunkIndex)
    {
        File file = chunkFile(chunkIndex);
        if (!file.delete())
        {
            throw new IllegalStateException("Can't collect garbage. Chunk index file: " + file);
        }
        allocationsPerChunk.remove(chunkIndex);
    }

    private File getChunkFileByPosition(Position writePosition)
    {
        return chunkFile(writePosition.chunkIndex);
    }

    private void advanceWritePosition(int chunkPosition)
    {
        position = getWritePosition().advance(chunkPosition);
    }

    private Position getWritePosition()
    {
        if (position.inChunkPosition < chunkSizeThreshold)
        {
            return position;
        }
        for (int notAllocatedChunkIndex = 0; notAllocatedChunkIndex < 65536; notAllocatedChunkIndex++)
        {
            if (!allocationsPerChunk.containsKey(notAllocatedChunkIndex))
            {
                position = new Position(notAllocatedChunkIndex, 0);
                return position;
            }
        }
        throw new IllegalStateException("BigMap is full");
    }

    private void closeStreams(Closeable fileStream, Closeable objectStream)
    {
        if (objectStream != null)
        {
            closeStream(objectStream);
            return;
        }
        closeStream(fileStream);
    }

    public V remove(Object key)
    {
        V value = get(key);
        indexMap.remove(key);
        return value;
    }

    public void putAll(Map<? extends K, ? extends V> m)
    {
        //TODO
    }

    public void clear()
    {
        //TODO
    }

    public Set<K> keySet()
    {
        return indexMap.keySet();
    }

    public Collection<V> values()
    {
        return throwExpensiveOperation();
    }

    public Set<Entry<K, V>> entrySet()
    {
        return throwExpensiveOperation();
    }

    public void flush()
    {
        ObjectOutputStream objectOutputStream = null;
        try
        {
            objectOutputStream = createObjectOutputStream(new FileOutputStream(index));
            objectOutputStream.writeObject(indexMap);
            objectOutputStream.writeObject(allocationsPerChunk);
            objectOutputStream.writeObject(position);
        }
        catch (IOException e)
        {
            throw new RuntimeException(e);
        }
        finally
        {
            closeStream(objectOutputStream);
        }
    }

    private void read()
    {
        ObjectInputStream inputStream = null;
        try
        {
            inputStream = createObjectInputStream(new FileInputStream(index));
            indexMap = (TreeMap<K, Position>) inputStream.readObject();
            allocationsPerChunk = (Map<Integer, Integer>) inputStream.readObject();
            position = (Position) inputStream.readObject();
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }
        finally
        {
            closeStream(inputStream);
        }
    }

    File chunkFile(int chunkIndex)
    {
        File chunksDir = new File(index.getAbsolutePath() + ".chunks");
        chunksDir.mkdir();
        return new File(chunksDir, "chunk" + chunkIndex);
    }

    private static class Position implements Externalizable
    {
        private int chunkIndex;
        private int inChunkPosition;

        private Position(int chunkIndex, int inChunkPosition)
        {
            this.chunkIndex = chunkIndex;
            this.inChunkPosition = inChunkPosition;
        }

        private Position advance(int chunkPosition)
        {
            return new Position(chunkIndex, chunkPosition);
        }

        public void writeExternal(ObjectOutput out) throws IOException
        {
            out.writeInt(chunkIndex);
            out.writeInt(inChunkPosition);
        }

        public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
        {
            chunkIndex = in.readInt();
            inChunkPosition = in.readInt();
        }
    }
}
